//package com.veryUsedCar.mq;
//
//import com.veryUsedCar.data.Result;
//import com.veryUsedCar.service.DownPriceService;
//import io.swagger.annotations.Api;
//import lombok.SneakyThrows;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.apache.rocketmq.spring.core.RocketMQListener;
//import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.cloud.context.config.annotation.RefreshScope;
//import org.springframework.stereotype.Component;
//import org.springframework.web.bind.annotation.GetMapping;
//import org.springframework.web.bind.annotation.PostMapping;
//import org.springframework.web.bind.annotation.RequestMapping;
//import org.springframework.web.bind.annotation.RestController;
//
//import java.util.List;
//
///**
// * author  woo
// * date 2022/11/19 15:31
// * 描述：
// */
//@Component
//@RefreshScope
//@Slf4j
//@Api("收藏页面的降价提醒功能接口")
//public class DownPriceController implements RocketMQListener <Long>, RocketMQPushConsumerLifecycleListener {
//    // 第一种方式: 设置一个定时器,定时查询价格,如果查询结果小于以前的价格,则通过mq给用户发送消息降价提醒
//    // 第二种方式: 在商家修改价格时,比较现有价格和之前价格,如果现有价格低于以前的价格,则通过mq给注册用户发送消息
//
//
//    // 第二种方式: 先暂时使用第二种方式,因为协作开发,先写接收消息(即消费者),提供者在修改价格时设定
//    @SneakyThrows
//    @Override
//    public void onMessage(Long aLong) {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("BroadcastingConsumerGroup");
//        consumer.setNamesrvAddr("182.92.204.196:9876");
//        consumer.setInstanceName("BroadcastingConsumer");
//
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//
//        /**
//         * set to broadcast mode
//         * 广播消息-消息被所有的订阅者消费
//         */
//        consumer.setMessageModel(MessageModel.BROADCASTING);
//        consumer.subscribe("downPrice", "*");
//
//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//                for (MessageExt msg : msgs) {
//                    System.out.println("BroadcastingConsumer-1 Receive New Messages ,queueId:" +
//                            msg.getQueueId() + ", content:" + new String(msg.getBody()));
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
//
//        consumer.start();
//
//        System.out.printf("BroadcastingConsumer-1 Started.%n");
//    }
//
//
//    @Override
//    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
//
//    }
//
//}


package com.veryUsedCar.mq;
import cn.hutool.json.JSONUtil;
import com.veryUsedCar.entity.Conllection;
import com.veryUsedCar.service.CollcetionService;
import com.veryUsedCar.service.PersonService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.mail.Message;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import java.util.Date;
import java.util.List;
import java.util.Properties;

@Component
//消费者需要用注解配置topic主题 和消费者的组
@RocketMQMessageListener(topic = "ggw", consumerGroup = "ggw")
//消费者需要继承接口RocketMQListener<>可以通过泛型接受消费消息
//ack重试机制要继承RocketMQPushConsumerLifecycleListener
public class DownPriceController implements RocketMQListener<Long>, RocketMQPushConsumerLifecycleListener {
    @Autowired
    private CollcetionService collcetionService;
    @Autowired
    private PersonService personService;

    //继承RocketMQListener的方法
    @Override
    public void onMessage(Long l) {
        //继承RocketMQPushConsumerLifecycleListener的接口后这个方法就作废了
        System.out.println("接收到信息 : " + l);
    }

    //继承RocketMQPushConsumerLifecycleListener后的方法
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        //通过方法创建内部类内部类中写业务
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //list就是消费消息
                //单条消息直接下表0获取对象,可以通过对象获取更多属性
                MessageExt messageExt = list.get(0);
                //new string接受消息内容 如果不转string会是一个二进制字符串
                String msg = new String(messageExt.getBody());
                Long aLong = Long.valueOf(msg);
                //获取消息id
                String msgId = messageExt.getMsgId();
                //获取消息重试次数
                int reconsumeTimes = messageExt.getReconsumeTimes();
                //ctrl+alt+t  生成try
                try {
                    // 根据商品id 获取收藏该商品的用户id的集合
                    List<Long> list1 = collcetionService.selectPerson(aLong);
                    // 根据用户id查询到收藏该商品用户的手机号
                    List<String> phones =  personService.selectPhone(list1);
                    // 遍历得到的用户 email ,循环发送降价提醒邮件
                    for (int i = 0; i < phones.size(); i++) {
                        downPriceEmail(phones.get(i));
                    }
                    System.out.println("ack消息:" + msg);
                    List<Long> longs = JSONUtil.toList(msg, Long.class);

                } catch (Exception e) {
                    e.printStackTrace();
                    if (reconsumeTimes >= 3) {//如果重试次数大于3,没必要继续尝试返回成功,并且将消息放入死信队列
                        System.out.println("放入死信队列");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    System.out.println("ACK消息接受失败,请稍后重试");
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//发生异常返回重新消费
                }
                //消费成功返回success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    public  void downPriceEmail(String Email) throws Exception {
        Properties properties = new Properties();
        //设置主机,端口,传输协议,是否需要身份验证
        properties.setProperty("mail.smtp.host","smtp.qq.com");
        properties.setProperty("mail.smtp.port","25");
        properties.setProperty("mail.transport.protocol", "smtp");
        properties.setProperty("mail.smtp.auth","true");

        //1.创建 Session 对象,加载 Properties 对象
        Session session = Session.getInstance(properties);

        session.setDebug(true);

        //2.通过 Session 对象,得到 transport 对象
        Transport transport = session.getTransport();

        //3.使用邮箱的用户名和密码,连接邮箱服务器
        transport.connect("smtp.qq.com","770969074","ybwxcvcirzxlgcab");

        //4.设置 Massage 邮件对象
        Message message = new MimeMessage(session);

        //发送者
        message.setFrom(new InternetAddress("770969074@qq.com"));
        //接受者
        message.setRecipient(Message.RecipientType.TO,new InternetAddress(Email)); // 动态获取接收者

        message.setSubject("你收藏的商品降价了,速来下单吧");
        message.setSentDate(new Date());
        message.setText("内容发送成功");


        //5.发送邮件
        transport.sendMessage(message,message.getAllRecipients());

        //关闭资源
        transport.close();

    }

}

